/**
 * Copyright (C) 2010, 2011 Neofonie GmbH
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package eu.dicodeproject.twitterstream.sink;

import java.io.IOException;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import eu.dicodeproject.analysis.hbase.TweetCols;

import twitter4j.GeoLocation;
import twitter4j.HashtagEntity;
import twitter4j.Place;
import twitter4j.Status;
import twitter4j.User;
import twitter4j.UserMentionEntity;

/**
 * Class that stores tweets to HBase.
 */
public class HBaseTweetSink implements TweetSink {
  /** Logger for this class. */
  private static final Logger log = LoggerFactory.getLogger(HBaseTweetSink.class);

  /** Column families. */
  private byte[] tweetFamily = Bytes.toBytes("d");
  private byte[] userFamily = Bytes.toBytes("d");
  /** Tweet table name. */
  private String tweetTableName = "twittertweets";
  /** User table name. */
  private String userTableName = "twitterusers";
  /** Zookeeper Quorum **/
  private String zookeeperQuorum = "localhost";
  /** Zookeeper Port **/
  private int zookeeperPort = 2181;
  /** HBase configuration. */
  private Configuration configuration;

  private HTablePool hTablePool;

  /** DateFormat used to parse tweet's creation date */
  // private final DateFormat dateFormat = new
  // SimpleDateFormat("EEE, dd MMM yyyy HH:mm:ss +SSS");

  /**
   * If a {@link Configuration} was set, that instance will be used. Otherwise
   * {@link #init()} creates a new one using the values given by
   * {@link #setZookeeperPort(int)} and {@link #setZookeeperQuorum(String)}.
   * 
   * @throws IOException
   */
  @PostConstruct
  public void init() throws IOException {
    if (this.configuration == null) {
      this.configuration = new Configuration();
      this.configuration.clear();
      this.configuration.set("hbase.zookeeper.quorum", this.zookeeperQuorum);
      this.configuration.setInt("hbase.zookeeper.property.clientPort", this.zookeeperPort);
    }
    HBaseAdmin admin = new HBaseAdmin(this.configuration);

    initTable(this.tweetTableName, this.tweetFamily, admin);
    initTable(this.userTableName, this.userFamily, admin);

    this.hTablePool = new HTablePool(this.configuration, 100);
  }

  @PreDestroy
  public void shutdown() {
    this.hTablePool.closeTablePool(this.userTableName);
    this.hTablePool.closeTablePool(this.tweetTableName);
  }

  private void initTable(String tablename, byte[] family, HBaseAdmin admin) throws IOException {
    if (!this.checkTable(tablename, admin)) {
      HTableDescriptor desc = new HTableDescriptor(tablename);
      HColumnDescriptor colText = new HColumnDescriptor(family);
      desc.addFamily(colText);
      if (log.isInfoEnabled()) {
	StringBuilder message = new StringBuilder("Creating table ");
	message.append(tablename);
	message.append(" with column family " + Bytes.toString(family) + " as non were found.");
	log.info(message.toString());
      }
      admin.createTable(desc);
    }
  }

  private boolean checkTable(final String tablename, final HBaseAdmin admin) throws IOException {
    HTableDescriptor[] descs = admin.listTables();
    for (HTableDescriptor desc : descs) {
      if (tablename.equals(desc.getNameAsString())) {
	return true;
      }
    }
    return false;
  }

  private void add(final byte[] family, final byte[] col, final String value, Put targetPut) {
    if (value != null) {
      targetPut.add(family, col, Bytes.toBytes(value));
    }
  }

  private void add(final byte[] family, final byte[] col, final long value, Put targetPut) {
    targetPut.add(family, col, Bytes.toBytes(value));
  }

  private void add(final byte[] family, final byte[] col, final int value, Put targetPut) {
    targetPut.add(family, col, Bytes.toBytes(value));
  }

  private void add(final byte[] family, final byte[] col, final double value, Put targetPut) {
    targetPut.add(family, col, Bytes.toBytes(value));
  }

  /** {@inheritDoc} */
  // TODO > 130 loc => time for some engineering.
  @Override
  public void store(final Status status) throws IOException {
    HTableInterface tweetTable = this.hTablePool.getTable(this.tweetTableName);
    HTableInterface userTable = this.hTablePool.getTable(this.userTableName);
    try {
      // we re-use the original tweet id from twitter as uuid
      // for more info on their generation implementation see
      // https://github.com/twitter/snowflake
      Put putTweet = new Put(Bytes.toBytes(status.getId()));
      Put putUser = null;
      // format as in original tweet Fri, 17 Dec 2010 20:58:49 +0000
      add(tweetFamily, TweetCols.SOURCE.bytes(), status.getSource(), putTweet);
      add(tweetFamily, TweetCols.TO.bytes(), "" + status.getInReplyToUserId(), putTweet);
      add(tweetFamily, TweetCols.TEXT.bytes(), status.getText(), putTweet);

      if (status.getCreatedAt() != null) {
	add(tweetFamily, TweetCols.CREATION_DATE.bytes(), status.getCreatedAt().getTime(), putTweet);
      }

      if (status.getGeoLocation() != null) {
	add(tweetFamily, TweetCols.GEO.bytes(), status.getGeoLocation().toString(), putTweet);
      }

      if (status.getUser() != null) {
	User user = status.getUser();

	// create put using twitter's user id as row key.
	// twitter's user ids from search api != user ids from twitter's rest
	// api...
	putUser = new Put(Bytes.toBytes(user.getId()));
	add(tweetFamily, TweetCols.FROM.bytes(), user.getName(), putTweet);
	add(userFamily, TweetCols.USER_NAME.bytes(), user.getName(), putUser);
	add(tweetFamily, TweetCols.FROM_ID.bytes(), user.getId(), putTweet); // long

	if (user.getProfileImageURL() != null) {
	  add(userFamily, TweetCols.IMAGE_URL.bytes(), user.getProfileImageURL().toString(), putUser);
	}

	add(tweetFamily, TweetCols.LANG.bytes(), user.getLang(), putTweet);
	add(userFamily, TweetCols.LANG.bytes(), user.getLang(), putUser);

	if (user.getCreatedAt() != null) {
	  add(userFamily, TweetCols.USER_CREATED_AT.bytes(), user.getCreatedAt().getTime(), putUser); // long
	}

	add(userFamily, TweetCols.USER_DESCRIPTION.bytes(), user.getDescription(), putUser);

	add(userFamily, TweetCols.USER_FAVOURITES_COUNT.bytes(), user.getFavouritesCount(), putUser); // int
	add(userFamily, TweetCols.USER_FOLLOWERS_COUNT.bytes(), user.getFollowersCount(), putUser); // int
	add(userFamily, TweetCols.USER_FRIENDS_COUNT.bytes(), user.getFriendsCount(), putUser); // int
	if (user.getListedCount() > 0) {
	  add(userFamily, TweetCols.USER_LISTED_COUNT.bytes(), user.getListedCount(), putUser); // int
	}
	add(userFamily, TweetCols.USER_LOCATION.bytes(), user.getLocation(), putUser);
	add(userFamily, TweetCols.USER_SCREEN_NAME.bytes(), user.getScreenName(), putUser);
	add(userFamily, TweetCols.USER_PROFILE_BACKGROUND_COLOR.bytes(), user.getProfileBackgroundColor(), putUser);
	add(userFamily, TweetCols.USER_PROFILE_BACKGROUND_IMAGE_URL.bytes(), user.getProfileBackgroundImageUrl(),
	    putUser);
	if (user.getProfileImageURL() != null) {
	  add(userFamily, TweetCols.USER_PROFILE_IMAGE_URL.bytes(), user.getProfileImageURL().toString(), putUser);
	}
	add(userFamily, TweetCols.USER_PROFILE_LINK_COLOR.bytes(), user.getProfileLinkColor(), putUser);
	add(userFamily, TweetCols.USER_PROFILE_SIDEBAR_BORDER_COLOR.bytes(), user.getProfileSidebarBorderColor(),
	    putUser);
	add(userFamily, TweetCols.USER_PROFILE_SIDEBAR_FILL_COLOR.bytes(), user.getProfileSidebarFillColor(), putUser);
	add(userFamily, TweetCols.USER_PROFILE_TEXT_COLOR.bytes(), user.getProfileTextColor(), putUser);
	add(userFamily, TweetCols.USER_STATUSES_COUNT.bytes(), user.getStatusesCount(), putUser); // int
	add(userFamily, TweetCols.USER_TIMEZONE.bytes(), user.getTimeZone(), putUser);
	if (user.getURL() != null) {
	  add(userFamily, TweetCols.USER_URL.bytes(), user.getURL().toString(), putUser);
	}
	add(userFamily, TweetCols.USER_UTC_OFFSET.bytes(), user.getUtcOffset(), putUser); // int
      }

      if (status.getHashtagEntities() != null) {
	StringBuilder hashTagsBuilder = new StringBuilder();
	for (HashtagEntity tag : status.getHashtagEntities()) {
	  hashTagsBuilder.append('#');
	  hashTagsBuilder.append(tag.getText());
	}
	add(tweetFamily, TweetCols.HASHTAGS.bytes(), hashTagsBuilder.toString(), putTweet);
      }

      add(tweetFamily, TweetCols.IN_REPLY_TO_STATUS_ID.bytes(), "" + status.getInReplyToStatusId(), putTweet);
      add(tweetFamily, TweetCols.IN_REPLY_TO_USER_ID.bytes(), "" + status.getInReplyToUserId(), putTweet);

      if (status.getPlace() != null) {
	Place place = status.getPlace();
	add(tweetFamily, TweetCols.PLACE_COUNTRY_CODE.bytes(), place.getCountryCode(), putTweet);
	add(tweetFamily, TweetCols.PLACE_FULL_NAME.bytes(), place.getFullName(), putTweet);
	add(tweetFamily, TweetCols.PLACE_TYPE.bytes(), place.getPlaceType(), putTweet);

	if (place.getGeometryType() != null) {
	  if (place.getGeometryType().equals("Point")) {
	    GeoLocation[][] coordinates = place.getGeometryCoordinates();
	    add(tweetFamily, TweetCols.PLACE_LATITUDE.bytes(), coordinates[0][0].getLatitude(), putTweet); // double
	    add(tweetFamily, TweetCols.PLACE_LONGITUDE.bytes(), coordinates[0][0].getLongitude(), putTweet); // double
	  } // else do something smart with polygons or keep ignoring them
	}
      }
      if (status.getRetweetCount() > 0) {
	add(tweetFamily, TweetCols.RETWEET_COUNT.bytes(), status.getRetweetCount(), putTweet); // long
      }
      // will this ever happen?
      if (status.getRetweetedStatus() != null) {
	this.store(status.getRetweetedStatus());
      }

      UserMentionEntity[] userMentionEntities = status.getUserMentionEntities();
      if (userMentionEntities != null && userMentionEntities.length > 0) {
	StringBuilder userIds = new StringBuilder();
	StringBuilder userNames = new StringBuilder();
	for (UserMentionEntity entity : userMentionEntities) {
	  userIds.append('#');
	  userIds.append(entity.getId());
	  userNames.append('#');
	  userNames.append(entity.getName());
	}
	add(tweetFamily, TweetCols.USER_IDS_MENTIONED.bytes(), userIds.toString(), putTweet);
	add(tweetFamily, TweetCols.USER_NAMES_MENTIONED.bytes(), userNames.toString(), putTweet);
      }

      tweetTable.put(putTweet);
      if (putUser != null) {
	userTable.put(putUser);
      }
    } catch (Exception e) {
      // we expect there to be parsing trouble and do not want to throw away the
      // whole batch of tweets in that case.
      log.warn(
	  "Error processing tweet: "
	      + e.getMessage()
	      + " (Ignoring and continuing processing - be alarmed if that happens often for your desired definition of often.) ",
	  e);
      log.warn("Tweet that let to the exception: " + status);
    } finally {
      this.hTablePool.putTable(userTable);
      this.hTablePool.putTable(tweetTable);
    }
  }

  /**
   * @param tweetFamily
   *          the tweetFamily to set
   */
  public void setTweetFamily(String tweetFamily) {
    this.tweetFamily = tweetFamily.getBytes();
  }

  /**
   * @param userFamily
   *          the userFamily to set
   */
  public void setUserFamily(String userFamily) {
    this.userFamily = userFamily.getBytes();
  }

  /**
   * @param tweetTableName
   *          the tweetTableName to set
   */
  public void setTweetTableName(String tweetTableName) {
    this.tweetTableName = tweetTableName.trim();
  }

  /**
   * @param userTableName
   *          the userTableName to set
   */
  public void setUserTableName(String userTableName) {
    this.userTableName = userTableName.trim();
  }

  /**
   * @param zookeeperQuorum
   *          the zookeeperQuorum to set
   */
  public void setZookeeperQuorum(String zookeeperQuorum) {
    this.zookeeperQuorum = zookeeperQuorum;
  }

  /**
   * @param zookeeperPort
   *          the zookeeperPort to set
   */
  public void setZookeeperPort(int zookeeperPort) {
    this.zookeeperPort = zookeeperPort;
  }

  /**
   * @param configuration
   *          the configuration to set
   */
  void setHBaseConfiguration(Configuration configuration) {
    this.configuration = configuration;
  }
}
